package com.navinfo.opentsp.platform.computing.serializer;

import com.alibaba.fastjson.JSON;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import org.apache.commons.lang.ArrayUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.sink.hbase.AsyncHbaseEventSerializer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.schema.types.*;
import org.hbase.async.AtomicIncrementRequest;
import org.hbase.async.PutRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * 驾驶行为分析2.0 D009协议异步序列化
 */
public class AsyncHbaseDrivingincidentAlarmSerializer implements AsyncHbaseEventSerializer {
    private static final Logger log = LoggerFactory.getLogger(AsyncHbaseDrivingincidentAlarmSerializer.class);
    private byte[] table;
    private byte[] colFam = Bytes.toBytes("0");
    private Event currentEvent;

    @Override
    public void initialize(byte[] table, byte[] colFam) {
        this.table = table;
        this.colFam = colFam;
    }

    @Override
    public void setEvent(Event event) {
        this.currentEvent = event;
    }

    @Override
    public void configure(Context context) {
    }
    @Override
    public void configure(ComponentConfiguration componentConfiguration) {
    }
    @Override
    public void cleanUp() {    }

    @Override
    public List<PutRequest> getActions() {
        List<PutRequest> actions = new ArrayList();
        try {
            LCIncidentAlarmData.IncidentAlarmData incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.parseFrom(currentEvent.getBody());


            //rowkey:reverse(tId)|3000000000L-gpsTime
            //reverse(tId)实现rowkey的散列，3000000000L-gpsTime实现逆序最新记录在头部
            //落盘phoenix时，phoenix主键 reverse(tId)+day+3000000000L-beginTime.
            //phoenix主键分隔符为\x00，所以各个主键字段分隔符为16进制的 0 （即\x00）
            String rowKey = new StringBuilder(Long.valueOf(incidentAlarmData.getTerminalId()).toString()).reverse() +"\\x00"+String.valueOf(DateUtils.format(incidentAlarmData.getGpsTime()*1000, "yyyyMMdd"))+"\\x00" +(3000000000L - incidentAlarmData.getGpsTime());
            String tidreverse = new StringBuilder(Long.valueOf(incidentAlarmData.getTerminalId()).toString()).reverse().toString();
            byte[] tid = Bytes.toBytes(tidreverse);
            byte sep = (byte) 0;
            byte[] seperate = new byte[] {sep};
            byte[] rowKeyByte = ArrayUtils.addAll(tid, seperate);
            byte[] day = Bytes.toBytes(String.valueOf(DateUtils.format(incidentAlarmData.getGpsTime()*1000, "yyyyMMdd")));
            byte[] rowKeyByte1 = ArrayUtils.addAll(rowKeyByte,day);
            byte[] rowKeyByte2 = ArrayUtils.addAll(rowKeyByte1,seperate);
            byte[] startTime = Bytes.toBytes(String.valueOf(3000000000L - incidentAlarmData.getGpsTime()));
            byte[] rowKeyByte3 = ArrayUtils.addAll(rowKeyByte2,startTime);
            byte[] serialNumber = Bytes.toBytes(String.valueOf(incidentAlarmData.getSerialNumber()));
            byte[] rowKeyByte4 = ArrayUtils.addAll(rowKeyByte3,seperate);
            byte[] rowKeyByte5= ArrayUtils.addAll(rowKeyByte4, serialNumber);


            //添加PutRequest
            actions.add(preparedata(rowKeyByte5 , incidentAlarmData));
        } catch (Exception e) {
            log.error("驾驶行为分析2.0 D009协议异步序列化失败：" + e.getMessage(), e);
        }

        return actions;
    }

    /**
     * 处理数据
     * @param rowKey
     * @throws Exception
     */
    private PutRequest preparedata(byte[] rowKey,  LCIncidentAlarmData.IncidentAlarmData incidentAlarmData ) throws Exception {

        byte[][] qualifiers = new byte[33][];
        byte[][] values = new byte[33][];

        qualifiers[0] = Bytes.toBytes("TID");
        qualifiers[1] = Bytes.toBytes("GPSTIME")                        ;
        qualifiers[2] = Bytes.toBytes("DAY");
        qualifiers[3] = Bytes.toBytes("SERIAL_NUMBER")  ;
        qualifiers[4] = Bytes.toBytes("RECEIVE_DATE")                   ;
        qualifiers[5] = Bytes.toBytes("LATITUDE")                       ;
        qualifiers[6] = Bytes.toBytes("LONGITUDE")                      ;
        qualifiers[7] = Bytes.toBytes("ORIGINALLNG")                    ;
        qualifiers[8] = Bytes.toBytes("ORIGINALLAT")                    ;
        qualifiers[9] = Bytes.toBytes("HEIGHT")                         ;
        qualifiers[10] =Bytes.toBytes("DIRECTION")                      ;
        qualifiers[11] =Bytes.toBytes("DEPOTLOGO")                      ;
        qualifiers[12] =Bytes.toBytes("RAPID_ACCELERATION_DURATION")    ;
        qualifiers[13] =Bytes.toBytes("RAPID_ACCELERATION_FUEL")        ;
        qualifiers[14] =Bytes.toBytes("RAPID_ACCELERATION_STATUS")      ;
        qualifiers[15] =Bytes.toBytes("RAPID_DECELERATION_DURATION")    ;
        qualifiers[16] =Bytes.toBytes("RAPID_DECELERATION_STATUS")      ;
        qualifiers[17] =Bytes.toBytes("OVER_SPEED_FUEL_CONSUMPTION")    ;
        qualifiers[18] =Bytes.toBytes("OVER_SPEED_STATUS")              ;
        qualifiers[19] =Bytes.toBytes("OVER_SPEED_DURATION")            ;
        qualifiers[20] =Bytes.toBytes("SUPER_LONG_IDLESTATUS")          ;
        qualifiers[21] =Bytes.toBytes("SUPER_LONG_IDLEDURATION")        ;
        qualifiers[22] =Bytes.toBytes("SUPER_LONG_IDLEFUEL_CONSUMPTION");
        qualifiers[23] =Bytes.toBytes("IDLE_SPEED_STATUS")              ;
        qualifiers[24] =Bytes.toBytes("IDLE_SPEED_DURATION")            ;
        qualifiers[25] =Bytes.toBytes("IDLE_SPEED_FUEL")                ;
        qualifiers[26] =Bytes.toBytes("BIG_THROTTLE_STATUS")            ;
        qualifiers[27] =Bytes.toBytes("BIG_THROTTLE_DURATION")          ;
        qualifiers[28] =Bytes.toBytes("BIG_THROTTLE_FUEL")              ;
        qualifiers[29] =Bytes.toBytes("NEUTRAL_TAXI_STATUS")            ;
        qualifiers[30] =Bytes.toBytes("NEUTRAL_TAXI_DURATION")          ;
        qualifiers[31] =Bytes.toBytes("FATIGUE_DRIVING_TIME")           ;
        qualifiers[32] =Bytes.toBytes("FATIGUE_DRIVING_STATUS")         ;


        values[0] = Bytes.toBytes(String.valueOf(incidentAlarmData.getTerminalId()));
        values[1] = Bytes.toBytes(String.valueOf(DateUtils.format(incidentAlarmData.getGpsTime())));
        values[2] = Bytes.toBytes(DateUtils.format(incidentAlarmData.getGpsTime()*1000, "yyyyMMdd"));
        values[3] = Bytes.toBytes(String.valueOf(incidentAlarmData.getSerialNumber()));
        values[4] = Bytes.toBytes(String.valueOf(incidentAlarmData.getReceiveDate()));
        values[5] = Bytes.toBytes(String.valueOf(incidentAlarmData.getLatitude()));
        values[6] = Bytes.toBytes(String.valueOf(incidentAlarmData.getLongitude()));
        values[7] = Bytes.toBytes(String.valueOf(incidentAlarmData.getOriginalLat()));
        values[8] = Bytes.toBytes(String.valueOf(incidentAlarmData.getOriginalLng()));
        values[9] = Bytes.toBytes(String.valueOf(incidentAlarmData.getHeight()));
        values[10] =Bytes.toBytes(String.valueOf(incidentAlarmData.getDirection()));
        values[11] =Bytes.toBytes(String.valueOf(incidentAlarmData.getDepotLogo()));
        values[12] =Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getDuration()));
        values[13] =Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getRapidAccelerationFuelConsumption()));
        values[14] =Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidAcceleration().getRapidStatus()));
        values[15] =Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidDeceleration().getDuration()));
        values[16] =Bytes.toBytes(String.valueOf(incidentAlarmData.getRapidDeceleration().getRapidDecelerationStatus()));
        values[17] =Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getFuelConsumption()));
        values[18] =Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getOverSpeedStatus()));
        values[19] =Bytes.toBytes(String.valueOf(incidentAlarmData.getOverSpeed().getDuration()));
        values[20] =Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getSuperLongIdleStatus()));
        values[21] =Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getDuration()));
        values[22] =Bytes.toBytes(String.valueOf(incidentAlarmData.getSuperLongIdle().getCumulativeFuelConsumption()));
        values[23] =Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getIdleSpeedAirConditionerStatus()));
        values[24] =Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getDuration()));
        values[25] =Bytes.toBytes(String.valueOf(incidentAlarmData.getIdleSpeedAirConditioner().getCumulativeFuelConsumption()));
        values[26] =Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getBigThrottleStatus()));
        values[27] =Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getDuration()));
        values[28] =Bytes.toBytes(String.valueOf(incidentAlarmData.getBigThrottle().getCumulativeFuelConsumption()));
        values[29] =Bytes.toBytes(String.valueOf(incidentAlarmData.getNeutralTaxi().getNeutralTaxiStatus()));
        values[30] =Bytes.toBytes(String.valueOf(incidentAlarmData.getNeutralTaxi().getDuration()));
        values[31] =Bytes.toBytes(String.valueOf(incidentAlarmData.getFatigueDriving().getFatigueDrivingTime()));
        values[32] =Bytes.toBytes(String.valueOf(incidentAlarmData.getFatigueDriving().getFatigueDrivingStatus()));



        return new PutRequest(table, rowKey, colFam, qualifiers, values);
    }

    @Override
    public List<AtomicIncrementRequest> getIncrements() {
        return new ArrayList<AtomicIncrementRequest>();
    }

    public static byte[] encodeDecimal(String value)
    {
        BigDecimal bigDecValue = new BigDecimal(value);
        byte[] pDecimalByteArray = PDecimal.INSTANCE.toBytes(bigDecValue);
        return pDecimalByteArray;
    }

    public static byte[] encodeDate(Date utilDate)
    {
        java.sql.Date sqlDate = new java.sql.Date(utilDate.getTime());
        byte[] pDateByteArray = PDate.INSTANCE.toBytes(sqlDate);
        return pDateByteArray;
    }

    public static byte[] encodeTimestamp(String utilTS)
    {
        Timestamp sqlDate = Timestamp.valueOf(utilTS);
        byte[] pTSByteArray = PTimestamp.INSTANCE.toBytes(sqlDate);
        return pTSByteArray;
    }

    public static byte[] encodeLong(String value)
    {
        byte[] pLongByteArray = PLong.INSTANCE.toBytes(value);
        return pLongByteArray;
    }

    public static byte[] encodeChar(String value)
    {
        byte[] pCharByteArray = PChar.INSTANCE.toBytes(value);
        return pCharByteArray;
    }

    public static byte[] encodeVarchar(String value)
    {
        byte[] pVarcharByteArray = PVarchar.INSTANCE.toBytes(value);
        return pVarcharByteArray;
    }

    public static byte[] encodeShort(String value)
    {
        Short shortValue = Short.parseShort(value);
        byte[] pShortByteArray = PSmallint.INSTANCE.toBytes(shortValue);
        return pShortByteArray;
    }

    public static byte[] encodeInteger(String value)
    {
        int intValue = Integer.parseInt(value);
        byte[] pIntByteArray = PInteger.INSTANCE.toBytes(intValue);
        return pIntByteArray;
    }
    
}